package net.bwie.realtime.vehicle.dwd.job.one;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import net.bwie.realtime.utils.DateTimeUtil;
import net.bwie.realtime.utils.DorisUtil;
import net.bwie.realtime.utils.KafkaUtil;
import net.bwie.realtime.vehicle.dwd.bean.MonitorInfo;
import net.bwie.realtime.vehicle.dwd.bean.MonitorInfoPair;
import net.bwie.realtime.vehicle.dwd.utils.AreaInfoUtil;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

/**
 {"actionTime": 1756260872, "monitorId": "M0291", "cameraId": "1", "car": "皖J73691", "speed": 55, "roadId": "R081", "areaId": "04", "longitude": 117.10191512979966, "latitude": 40.108698061038886}
 {"actionTime": 1756260878, "monitorId": "M0387", "cameraId": "1", "car": "皖J73691", "speed": 55, "roadId": "R081", "areaId": "04", "longitude":116.38736097191126,"latitude":39.86497137051134}
 {"actionTime": 1756260881, "monitorId": "M0385", "cameraId": "2", "car": "贵S03745", "speed": 100, "roadId": "R107", "areaId": "12", "longitude": 116.38642859280465, "latitude": 39.86749424661375}
 {"actionTime": 1756260886, "monitorId": "M0387", "cameraId": "2", "car": "贵S03745", "speed": 170, "roadId": "R107", "areaId": "12", "longitude":116.38736097191126,"latitude":39.86497137051134}
 {"actionTime": 1756260887, "monitorId": "M0386", "cameraId": "2", "car": "贵S03745", "speed": 10, "roadId": "R107", "areaId": "12", "longitude":116.38736097191626,"latitude":39.86497137041134}

 {"actionTime": 1756260986, "monitorId": "M9999", "cameraId": "2", "car": "===S999999", "speed": 99, "roadId": "R107", "areaId": "12", "longitude":116.38736097199126,"latitude":39.86497137051134}
 {"actionTime": 1756260987, "monitorId": "M8888", "cameraId": "2", "car": "===S999999", "speed": 88, "roadId": "R107", "areaId": "12", "longitude":116.38736097191126,"latitude":39.86497137051134}

 {"actionTime": 1756260989, "monitorId": "M0386", "cameraId": "2", "car": "贵S999999", "speed": 170, "roadId": "R107", "areaId": "12", "longitude":116.4084975177547,"latitude":39.91545432827133}
 {"actionTime": 1756260990, "monitorId": "M0385", "cameraId": "2", "car": "贵S999999", "speed": 170, "roadId": "R107", "areaId": "12", "longitude":116.10191512979966,"latitude":39.108698061038886}
 {"actionTime": 1756260991, "monitorId": "M0385", "cameraId": "2", "car": "贵S999999", "speed": 170, "roadId": "R107", "areaId": "12", "longitude":116.10191512979966,"latitude":39.108698061038886}

 *
 * @author ASUS*/
/*todo 2 套牌分析
 * 当某个卡口中出现一辆行驶的汽车，我们可以通过摄像头识别车牌号，然后在10秒内，另
 * 外一个卡口（或者当前卡口）也识别到了同样车牌的车辆，那么很有可能这两辆车之中有很大几率存在套牌车，
 * 因为一般情况下不可能有车辆在10秒内经过两个卡口。如果发现涉嫌套牌车，系统实时发出报警信息，同时这些存在套牌车嫌疑的车辆，写入Mysql数据库的结果表中，在后面的模块中，可以对这些违法车辆进行实时轨迹跟踪。
 * */
public class DeckAnalysisJob {

    public static void main(String[] args) throws Exception {

        //环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        // 增加网络和检查点配置
        env.setParallelism(1);
        // 增加检查点间隔
        env.enableCheckpointing(5000L);
        // 设置检查点超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        //读取主题
        DataStream<String> vehicleKafkaStream = KafkaUtil.consumerKafka(env, "traffic_monitor_data");

//        vehicleKafkaStream.print();

        DataStream<String> stringDataStream = processJson(vehicleKafkaStream);
        //处理
        DataStream<String> result = mapWindowData(stringDataStream);

        result.print();
        //4. sink-数据输出
        DorisUtil.saveToDoris(result,
                "vehicle_realtime_report",
                "vehicle_deck_analy_violation");


        //5. execute-执行
        env.execute("RealtimeDeckAnalysisJob");
    }

    private static DataStream<String> processJson(DataStream<String> vehicleKafkaStream) {
        //转为实体类，并对时间戳进行毫秒级的转化
        SingleOutputStreamOperator<String> mapStream = vehicleKafkaStream.map(new MapFunction<String, String>() {
            @Override
            public String map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                jsonObject.put("actionTime", jsonObject.getLong("actionTime") * 1000);
                return jsonObject.toJSONString();
            }
        });


        // 3. 调用工具类预加载区域信息
        AreaInfoUtil.preloadAreaInfo();

        // 4. 处理JSON数据（调用工具类的方法）


        return AreaInfoUtil.processJson(mapStream);
    }

    private static DataStream<String> mapWindowData(DataStream<String> vehicleKafkaStream) {


        //转为实体类，并对时间戳进行毫秒级的转化
        SingleOutputStreamOperator<MonitorInfo> monitorinfo = vehicleKafkaStream.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String s) throws Exception {
                return JSON.parseObject(s, MonitorInfo.class);
            }
        });

//        monitorinfo.print();

        // 2. 解析JSON后，提取事件时间并生成水位线
        SingleOutputStreamOperator<MonitorInfo> withWatermarkDs = monitorinfo
                // 提取事件时间（从MonitorInfo的actionTime字段获取，单位：毫秒）
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy

                                .<MonitorInfo>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner((event, timestamp) -> event.getActionTime())
                );
//        withWatermarkDs.print();

// 定义侧输出标签（用于Kafka输出）
// 定义侧输出标签（必须是静态的）
        OutputTag<String> kafkaOutputTag = new OutputTag<String>("kafka-events") {};

// 2. 用ProcessFunction替代flatMap，处理逻辑更清晰
        SingleOutputStreamOperator<Violation> resultDs = withWatermarkDs
                .keyBy(MonitorInfo::getCar)
                .process(new ProcessFunction<MonitorInfo, Violation>() {

                    private static final long TIME_THRESHOLD = 10000; // 10秒阈值
                    private ListState<MonitorInfo> eventListState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 初始化状态（保存同一车辆的事件）
                        ListStateDescriptor<MonitorInfo> stateDesc = new ListStateDescriptor<>(
                                "carEventList",
                                MonitorInfo.class
                        );

                        eventListState = getRuntimeContext().getListState(stateDesc);

                    }

                    @Override
                    public void processElement(MonitorInfo newEvent, Context ctx, Collector<Violation> out) throws Exception {

                        // 1. 新事件加入状态
                        eventListState.add(newEvent);

                        // 2. 取出所有事件并按时间升序排序（最早→最新）
                        List<MonitorInfo> eventList = new ArrayList<>();
                        for (MonitorInfo event : eventListState.get()) {
                            eventList.add(event);
                        }
                        eventList.sort(Comparator.comparingLong(MonitorInfo::getActionTime));

                        // 3. 筛选10秒内有效事件 + 同时保留最近2条（核心逻辑）
                        List<MonitorInfo> validAndLatest2Events = new ArrayList<>();
                        long newEventTime = newEvent.getActionTime();
                        for (MonitorInfo event : eventList) {
                            // 第一步：判断是否在10秒内（有效事件）
                            if (newEventTime - event.getActionTime() <= TIME_THRESHOLD) {
                                // 第二步：若有效事件已超过2条，移除最早的1条（保持窗口大小为2）
                                if (validAndLatest2Events.size() >= 2) {
                                    validAndLatest2Events.remove(0); // 移除索引0（最早的有效事件）
                                }
                                // 第三步：将当前有效事件加入列表（始终在末尾，保证列表是“较新→最新”）
                                validAndLatest2Events.add(event);
                            }
                            // 无效事件（超过10秒）直接跳过，不加入列表
                        }

                        // 4. 检测套牌行为，输出结果 3

                        for (int i = validAndLatest2Events.size()-1; i >0; i--) {
                            MonitorInfo prevEvent = validAndLatest2Events.get(i-1);
                            MonitorInfo currEvent = validAndLatest2Events.get(i);
                            // 判定条件：10秒内且监控点不同
                            if (currEvent.getActionTime() - prevEvent.getActionTime() <= TIME_THRESHOLD
                                    &&currEvent.getActionTime() - prevEvent.getActionTime()>0
                                    && !currEvent.getMonitorId().equals(prevEvent.getMonitorId())) {

                                // a. 主输出：Violation实体（存数据库）

                                Violation violation = new Violation(
                                        currEvent.getCar(),
                                        prevEvent.getAreaName(),
                                        prevEvent.getMonitorId(),
                                        DateTimeUtil.convertLongToString(currEvent.getActionTime(), DateTimeUtil.DATE_TIME_FORMAT),
                                        DateTimeUtil.convertLongToString(prevEvent.getActionTime(), DateTimeUtil.DATE_TIME_FORMAT),
                                        (currEvent.getActionTime() - prevEvent.getActionTime()) / 1000,
                                        PositionUtil.getDistance1(
                                                currEvent.getLongitude(), currEvent.getLatitude(),
                                                prevEvent.getLongitude(), prevEvent.getLatitude()
                                        )
                                );
                                out.collect(violation); // 主输出流

                                // b. 侧输出：通过Context发送prevEvent和currEvent（发Kafka）
                                // 1. 组装成对数据：将 prevEvent 和 currEvent 封装到 MonitorInfoPair 中
                                MonitorInfoPair monitorPair = new MonitorInfoPair(prevEvent, currEvent);

                                // 2. 序列化为 JSON 字符串（FastJSON 自动将对象转为指定格式的 JSON）
                                String jsonOutput = JSON.toJSONString(monitorPair);

                                // 3. 侧输出到 Kafka（此时输出的是标准 JSON 字符串）
                                ctx.output(kafkaOutputTag, jsonOutput);

                                System.out.println("==================================");
                            }
                        }


                    }
                });

//        resultDs.print();
// 转换为 JSON 字符串
        DataStream<String> jsonStream = resultDs.map(violation -> {
            // 处理null值，避免JSON格式错误
            String car = violation.getCar() != null ? violation.getCar() : "";
            String firstActionTime = violation.getFirst_action_time() != null ? violation.getFirst_action_time() : "";
            String secondActionTime = violation.getSecond_action_time() != null ? violation.getSecond_action_time() : "";
            long interval = violation.getInterval(); // long类型默认0，无需额外处理
            double distance = violation.getDistance(); // double类型默认0.0，无需额外处理
            String areaName = violation.getAreaName() != null ? violation.getAreaName() : "";
            String monitorId = violation.getMonitorId() != null ? violation.getMonitorId() : "";
            // 构建JSON字符串（包含所有字段）
            return String.format(
                    "{\"car\":\"%s\",\"area_name\":\"%s\",\"monitor_id\":\"%s\",\"first_action_time\":\"%s\",\"second_action_time\":\"%s\",\"interval\":%d,\"distance\":%.2f}",
                    car,
                    areaName,
                    monitorId,
                    firstActionTime,
                    secondActionTime,
                    interval,
                    distance
            );
        });

        KafkaUtil.producerKafka( resultDs.getSideOutput(kafkaOutputTag), "dws_deck");

        return jsonStream;
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Violation {
        private String car;
        private String areaName;
        private String monitorId;
        private String first_action_time;
        private String second_action_time;
        private long   interval;
        private double distance;
    }
}

